package com.atguigu.day02;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Example6 {
    public static void main(String[] args) throws Exception {
        var sourceToMapQueue = new ArrayBlockingQueue<String>(10);
        var mapToReduceQueue = new ArrayBlockingQueue<Tuple2<String, Integer>>(10);

        var source = new SourceFunction(sourceToMapQueue);
        var map = new MapFunction(sourceToMapQueue, mapToReduceQueue);
        var reduce = new ReduceFunction(mapToReduceQueue);

        var t1 = new Thread(source);
        var t2 = new Thread(map);
        var t3 = new Thread(reduce);

        t1.start();
        t2.start();
        t3.start();

        t1.join();
        t2.join();
        t3.join();
    }

    public static class SourceFunction implements Runnable {
        private final BlockingQueue<String> queue;

        public SourceFunction(BlockingQueue<String> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            queue.add("hello world");
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            queue.add("hello world");
        }
    }

    public static class MapFunction implements Runnable {
        private final BlockingQueue<String> source;
        private final BlockingQueue<Tuple2<String, Integer>> target;

        public MapFunction(BlockingQueue<String> source, BlockingQueue<Tuple2<String, Integer>> target) {
            this.source = source;
            this.target = target;
        }

        @Override
        public void run() {
            while (true) {
                var value = source.poll();
                if (value != null) {
                    var words = value.split(" ");
                    for (var word : words) {
                        target.add(Tuple2.of(word, 1));
                    }
                }
            }
        }
    }

    public static class ReduceFunction implements Runnable {
        private final BlockingQueue<Tuple2<String, Integer>> source;
        private final Map<String, Tuple2<String, Integer>> state;

        public ReduceFunction(BlockingQueue<Tuple2<String, Integer>> source) {
            this.source = source;
            this.state = new HashMap<>();
        }

        @Override
        public void run() {
            while (true) {
                var value = source.poll();
                if (value != null) {
                    if (state.get(value.f0) == null) {
                        state.put(value.f0, Tuple2.of(value.f0, 1));
                    } else {
                        var oldAcc = state.get(value.f0);
                        var newAcc = Tuple2.of(oldAcc.f0, oldAcc.f1 + value.f1);
                        state.put(value.f0, newAcc);
                    }
                    System.out.println(state.get(value.f0));
                }
            }
        }
    }



    public static class Tuple2<T0, T1> {
        public T0 f0;
        public T1 f1;

        public Tuple2() {}

        public Tuple2(T0 f0, T1 f1) {
            this.f0 = f0;
            this.f1 = f1;
        }

        public static <T0, T1> Tuple2<T0, T1> of(T0 f0, T1 f1) {
            return new Tuple2<>(f0, f1);
        }

        @Override
        public String toString() {
            return "(" + f0 + ", " + f1 + ")";
        }
    }
}
